package com.chinaums.sftp.tablesource;

import com.chinaums.sftp.internal.options.SftpReadOptions;
import com.chinaums.sftp.source.SftpParallelSourceFunction;
import com.chinaums.sftp.source.SftpSourceFunction;
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.table.connector.format.DecodingFormat;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.connector.source.ScanTableSource;
import org.apache.flink.table.connector.source.SourceFunctionProvider;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.types.DataType;

import static com.chinaums.sftp.internal.constants.ConstantValue.SINGLE_STREAM;

import java.util.Objects;

/**
 * table source for ftp.
 */
@Internal
public class SftpDynamicTableSource implements ScanTableSource {
    private final SftpReadOptions sftpReadOptions;
    private final DecodingFormat<DeserializationSchema<RowData>> decodingFormat;
    private final DataType producedDataType;


    public SftpDynamicTableSource(
            SftpReadOptions sftpReadOptions,
            DecodingFormat<DeserializationSchema<RowData>> decodingFormat,
            DataType schema) {
        this.sftpReadOptions = sftpReadOptions;
        this.decodingFormat = decodingFormat;
        this.producedDataType = schema;
    }

    @Override
    public ChangelogMode getChangelogMode() {
        return decodingFormat.getChangelogMode();
    }

    @Override
    public ScanRuntimeProvider getScanRuntimeProvider(ScanContext runtimeProviderContext) {
        DeserializationSchema<RowData> runtimeDecoder = decodingFormat.createRuntimeDecoder(
                runtimeProviderContext,
                producedDataType
        );
        if (SINGLE_STREAM.equalsIgnoreCase(sftpReadOptions.getReadMode())) {
            SftpSourceFunction sftpSourceFunction = new SftpSourceFunction(
                    sftpReadOptions,
                    runtimeDecoder,
                    runtimeProviderContext.createTypeInformation(producedDataType));
            return SourceFunctionProvider.of(sftpSourceFunction, false);
        } else {
            SftpParallelSourceFunction sftpParallelSourceFunction = new SftpParallelSourceFunction(
                    sftpReadOptions,
                    runtimeDecoder,
                    runtimeProviderContext.createTypeInformation(producedDataType));
            return SourceFunctionProvider.of(sftpParallelSourceFunction, false);
        }

    }

    @Override
    public DynamicTableSource copy() {
        return new SftpDynamicTableSource(sftpReadOptions, decodingFormat, producedDataType);
    }

    @Override
    public String asSummaryString() {
        return "Sftp Table Source";
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) {
            return true;
        }
        if (!(o instanceof SftpDynamicTableSource)) {
            return false;
        }
        SftpDynamicTableSource that = (SftpDynamicTableSource) o;
        return Objects.equals(sftpReadOptions, that.sftpReadOptions) &&
                Objects.equals(decodingFormat, that.decodingFormat) &&
                Objects.equals(producedDataType, that.producedDataType);
    }


    @Override
    public int hashCode() {
        return Objects.hash(sftpReadOptions, decodingFormat, producedDataType);
    }
}
